package com.xxxx;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;


public class Hello02WaterMark {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //水印使用前提，基于（事件时间）时间窗口
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //间隔 1S
        env.getConfig().setAutoWatermarkInterval(1000);

        DataStreamSource<String> source = env.socketTextStream("192.168.88.101", 18880);

        SingleOutputStreamOperator<String> andWatermarks = source.assignTimestampsAndWatermarks(new PeriodicWaterMark());

        andWatermarks.map(word -> Tuple2.of(word.split(",")[0], 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .timeWindow(Time.seconds(10))
                .sum(1)
                .print();
        env.execute();
    }
}

class PeriodicWaterMark implements AssignerWithPeriodicWatermarks<String> {

    //数据允许的延迟情况
    long maxLateTime = 5000;
    //当前系统最大的时间
    long currentMaxTimestamp = Long.MIN_VALUE;


    //水印产生，周期性产生，默认200ms，基于自己业务的时间容忍度去产生水印，因为要通过水印来解决数据的延迟/乱序问题
    @Override
    public Watermark getCurrentWatermark() {
        long timestamp = System.currentTimeMillis() - maxLateTime;
        System.out.println("MyWaterMark.getCurrentWatermark[" + timestamp + maxLateTime * 2 + "]");
        //本次水位线的位置
        Watermark waterMark = new Watermark(timestamp);
        return waterMark;
    }

    /**
     * 从事件中抽取时间，假设数据格式为 hello,2020-12-31 11:59:33.830
     *
     * @param element
     * @param previousElementTimestamp
     * @return
     */
    @Override
    public long extractTimestamp(String element, long previousElementTimestamp) {
        long eventTimestamp = Long.valueOf(element.split(",")[1]);

        System.err.println("MyWaterMark.extractTimestamp事件信息[" + element + "]最新时间【" + (System.currentTimeMillis() - 5000) + "】");
        return eventTimestamp;
    }
}